package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import com.atguigu.tingshu.album.mapper.*;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumAttributeValueMapper albumAttributeValueMapper;

	@Autowired
	private AlbumStatMapper albumStatMapper;

	@Autowired
	private TrackInfoMapper trackInfoMapper;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private TrackStatMapper trackStatMapper;

	@Autowired
	private RedissonClient redissonClient;

	/**
	 * 保存专辑方法
	 * 1.将提交专辑VO转为PO对象，新增专辑
	 * 2.将提交专辑标签封装为专辑标签关系集合对象，进行批量保存
	 * 3.为新增专辑标签，批量保存专辑统计信息
	 *
	 * @param userId      用户ID
	 * @param albumInfoVo 专辑信息VO对象
	 */
	@Override
	@Transactional(rollbackFor = Exception.class) //默认事务注解修饰方法捕获运行时异常或者Error
	public void saveAlbumInfo(Long userId, AlbumInfoVo albumInfoVo) {
		//1.保存专辑
		//1.1 将提交专辑VO拷贝到专辑PO对象中 仅限同名同类型属性才会被拷贝
		AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);

		//1.2 手动为用户id、包含声音数量（0）、是否完结、免费试听集数、审核状态（TODO：审核通过，实则有内容审核机制）
		albumInfo.setUserId(userId);
		albumInfo.setIncludeTrackCount(0);
		albumInfo.setIsFinished("0");
		albumInfo.setTracksForFree(5);
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		//1.3 保存专辑，得到专辑ID
		albumInfoMapper.insert(albumInfo);
		Long albumId = albumInfo.getId();

		//2.保存专辑标签关系
		//2.1 将提交专辑标签关系VO集合转为PO集合
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
			//2.2 关联专辑ID批量保存专辑标签关系
			for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
				AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
				albumAttributeValue.setAlbumId(albumId);
				albumAttributeValueMapper.insert(albumAttributeValue);
			}

		}
		//3.保存专辑统计信息（4条）
		this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY, 0);
		this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE, 0);
		this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_BUY, 0);
		this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT, 0);

		//4.TODO 调用内容审核接口（第三方阿里云）对专辑内容（封面、文字）

		//5.审核通过后发送上架专辑消息到Kafka
		if ("1".equals(albumInfo.getIsOpen())) {
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
		}

		//将新增的专辑id存入布隆过滤器
		//获取布隆过滤器
		RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
//		将新增专辑ID存入布隆过滤器
		bloomFilter.add(albumInfo.getId());

	}

	/**
	 * 保存专辑统计信息
	 * @param albumId 专辑ID
	 * @param statType 统计类型
	 * @param statNum 统计数量
	 */
	@Override
	public void saveAlbumStat(Long albumId, String statType, int statNum) {
		AlbumStat albumStat = new AlbumStat();
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statType);
		albumStat.setStatNum(statNum);
		albumStatMapper.insert(albumStat);
	}

	/**
	 * 查询当前用户专辑分页列表
	 * @param pageInfo 分页对象
	 * @param albumInfoQuery 查询条件对象
	 * @return
	 */
	@Override
	public Page<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery albumInfoQuery) {


		return albumInfoMapper.getUserAlbumPage(pageInfo, albumInfoQuery);
	}

	/**
	 * 根据专辑ID删除专辑
	 * @param id
	 * @return
	 */
	@Override
	@Transactional(rollbackFor = Exception.class)
	public void removeAlbumInfo(Long id) {

		//1.根据专辑id查询该专辑下是否有声音，有声音不允许删除
		LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
		queryWrapper.eq(TrackInfo::getAlbumId, id);

		Long count = trackInfoMapper.selectCount(queryWrapper);
		if (count > 0){
			throw new GuiguException(400, "该专辑下有声音不许删除！");
		}

		//2.根据主键删除专辑
		albumInfoMapper.deleteById(id);

		//3.根据专辑ID删除专辑标签
		LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
		albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

		//4.根据专辑ID删除专辑统计数据
		LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
		albumStatMapper.delete(albumStatLambdaQueryWrapper);

		//TODO 同步删除索引库中专辑文档
		kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
	}


	/**
	 *
	 *  获取专辑信息（SpringDataRedis实现分布式锁）避免缓存击穿
	 *
	 * 获取专辑信息（Redissson实现分布式锁）避免缓存击穿
	 *
	 * @param id
	 * @return
	 */
	@Override
	@GuiGuCache(prefix = "album:info:")
	public AlbumInfo getAlbumInfo(Long id) {

//		获取专辑信息（Redissson实现分布式锁）避免缓存击穿
	//	return getAlbumInfoFromRedisson(id);

//		//获取专辑信息（SpringDataRedis实现分布式锁）避免缓存击穿
//		return getAlbumInfoFromRedis(id);
//
//		//从数据库获取专辑信息
		return getAlbumInfoFromDB(id);
	}



	/**
	 * 获取专辑信息（Redissson实现分布式锁）避免缓存击穿
	 *
	 * @param id
	 * @return
	 */
	private AlbumInfo getAlbumInfoFromRedisson(Long id) {
		try {
			//1.优先从缓存中获取数据
			//1.1 构建业务数据Key 形式：前缀+业务唯一标识
			String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
			//1.2 查询Redis获取业务数据
			AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
			//1.3 命中缓存则直接返回
			if (albumInfo != null) {
				log.info("命中缓存，直接返回，线程ID：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				return albumInfo;
			}
			//2.获取分布式锁
			//2.1 构建锁key
			String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
			//2.2 基于RedissonClient创建锁对象 入参成为Redis中Hash结构锁的key
			RLock lock = redissonClient.getLock(lockKey);
			//2.3 获取锁-阻塞直到获取锁成功-lock方法 入参释放锁时间-1（开启看门狗） 默认锁时间：30s
			lock.lock();

			try {
				//3.获取锁成功执行业务,将查询业务数据放入缓存Redis
				//3.1 处于阻塞等待获取线程（终将获取锁成功）避免获取锁线程再次查库，这里再查一次缓存

				/*redisson默认自旋是只在
				*
				* 	RLock lock = redissonClient.getLock(lockKey);
				*	lock.lock();
				* 不会从头开开始自旋，所以当有一个线程已经查完数据库数据存到redis中时，redisson并没有自旋到从redis中获取数据所以再查一次缓存
				*
				* */
				albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
				if (albumInfo != null) {
					log.info("当前线程{},获取锁成功，且再次命中缓存成功", Thread.currentThread().getName());
					return albumInfo;
				}
				//3.2 查询数据库将查询结果放入缓存Redis 返回
				albumInfo = this.getAlbumInfoFromDB(id);
				long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT;
				redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
				return albumInfo;
			} finally {
				//4.业务执行完毕释放锁
				log.info("当前线程：{}，释放锁", Thread.currentThread().getName());
				lock.unlock();
			}
		} catch (Exception e) {
			//5.兜底处理方案：Redis服务有问题，将业务数据获取自动从数据库获取
			log.error("[专辑服务]查询专辑数据异常：{}", e);
			return this.getAlbumInfoFromDB(id);
		}
	}


	@Override
	public AlbumInfo getAlbumInfoFromRedis(Long id){
		try {
			//1.优先从缓存中获取数据
			//1.1 构建业务数据Key 形式：前缀+业务唯一标识
			String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
			//1.2 查询Redis获取业务数据
			AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
			//1.3 命中缓存则直接返回
			if (albumInfo != null) {
				log.info("命中缓存，直接返回，线程ID：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				return albumInfo;
			}
			//2.尝试获取分布式锁（set k v ex nx可能获取锁失败）
			//2.1 构建锁key
			String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
			//2.2 采用UUID作为线程标识
			String lockVal = IdUtil.fastSimpleUUID();
			//2.3 利用Redis提供set nx ex 获取分布式锁
			Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockVal, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
			if (flag) {
				//3.获取锁成功执行业务,将查询业务数据放入缓存Redis
				log.info("获取锁成功：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				try {
					albumInfo = this.getAlbumInfoFromDB(id);
					long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT;
					redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
					return albumInfo;
				} finally {
					//4.业务执行完毕释放锁
					String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
							"then\n" +
							"    return redis.call(\"del\",KEYS[1])\n" +
							"else\n" +
							"    return 0\n" +
							"end";
					DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
					redisScript.setScriptText(scriptText);
					redisScript.setResultType(Long.class);
					redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockVal);
				}
			} else {
				try {
					//5.获取锁失败则自旋（业务要求必须执行）
					Thread.sleep(200);
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
				log.error("获取锁失败，自旋：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				return this.getAlbumInfo(id);
			}
		} catch (Exception e) {
			//兜底处理方案：Redis服务有问题，将业务数据获取自动从数据库获取
			log.info("[专辑服务]Redis服务异常：{}", e);
			return this.getAlbumInfoFromDB(id);
		}
	}

	/***
	 * 根据专辑ID查询专辑信息包含专辑标签列表
	 * @param id 专辑ID
	 * @return
	 */
	@Override
	public AlbumInfo getAlbumInfoFromDB(Long id) {
		//1.根据主键查询专辑信息
		AlbumInfo albumInfo = albumInfoMapper.selectById(id);

		//2.根据专辑ID查询专辑标签列表
		if (albumInfo != null) {
			LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
			queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
			List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
			albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
		}
		return albumInfo;
	}

	/**
	 * 修改专辑信息
	 * @param id 专辑ID
	 * @param albumInfoVo 修改专辑信息VO对象
	 * @return
	 */
	@Override
	@Transactional(rollbackFor = Exception.class)
	public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
		//1.修改专辑信息
		//1.1 将修改专辑信息转为PO对象 封装专辑ID
		AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
		albumInfo.setId(id);

		//1.2 修改专辑
		albumInfoMapper.updateById(albumInfo);

		//2.修改专辑标签 先删再新增
		//2.1 根据专辑ID将以前关联标签记录全部删除
		QueryWrapper<AlbumAttributeValue> queryWrapper = new QueryWrapper<>();
		queryWrapper.eq("album_id", id);
		albumAttributeValueMapper.delete(queryWrapper);

		//2.2 在根据提交专辑标签做新增
		List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();

		if (CollectionUtil.isEmpty(albumAttributeValueVoList)){
			throw new GuiguException(400, "专辑信息不完整，没有属性值！");
		}

		//2.2 关联专辑ID批量保存专辑标签关系
		for (AlbumAttributeValue albumAttributeValue : albumAttributeValueVoList) {
			albumAttributeValue.setAlbumId(id);
			albumAttributeValueMapper.insert(albumAttributeValue);
		}

		//TODO 如果是开放专辑自动将专辑同步到ES索引库中
		if ("1".equals(albumInfo.getIsOpen())) {
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfo.getId().toString());
		} else {
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumInfo.getId().toString());
		}

	}

	/**
	 * 查询当指定用户专辑列表
	 * @param userId
	 * @return
	 */
	@Override
	public List<AlbumInfo> getUserAllAlbumList(Long userId) {

		//select id, album_title from album_info where user_id = ? order by id desc limit 200
		//1.创建条件构造器封装查询条件
		LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
		queryWrapper.eq(AlbumInfo::getUserId, userId);
		//2.指定查询字段 select 后面列名称
		queryWrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle);
		//3.设置排序按照最新专辑
		queryWrapper.orderByDesc(AlbumInfo::getId);
		//4.TODO 避免小程序响应1千多条记录导致卡顿 需要指定加载数量  limit 200 -> 显示200条数据
		queryWrapper.last(" limit 200 "); //拼接sql语句在最后，加个空格，防止拼接直接拼到一起导致出错

		return albumInfoMapper.selectList(queryWrapper);
	}

	/**
	 * 根据专辑ID查询专辑统计信息
	 *
	 * @param albumId
	 * @return
	 */
	@Override
	@GuiGuCache(prefix = "albumStatVo:")
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		return albumInfoMapper.getAlbumStatVo(albumId);
	}

	/**
	 * MQ监听更新声音统计信息（包含：播放、收藏、点赞、评论）
	 *
	 * @param mqVo
	 */
	@Override
	@Transactional(rollbackFor = Exception.class)
	public void updateTrackStat(TrackStatMqVo mqVo) {

		//1.做幂等性处理，统一个消息只处理一次 采用set k(业务消息唯一标识) v NX EX
		String key = "mq:" + mqVo.getBusinessNo();

		try {
			//存活时间为  消息队列可能重复发送的时间间隔 设置存活为1小时
			Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, mqVo.getBusinessNo(), 1, TimeUnit.HOURS);

			if (flag){
				// 说明是第一次存
				//2 更新声音统计信息
				trackStatMapper.updateStat(mqVo.getTrackId(), mqVo.getStatType(), mqVo.getCount());

				//3.更新专辑统计信息(播放量、评论量只要声音+1，对应专辑也得+1) 声音类型0701 0704 而专辑对应的类型为0401 0404
				if (SystemConstant.TRACK_STAT_PLAY.equals(mqVo.getStatType())) {
					albumStatMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_PLAY, mqVo.getCount());
				}

				if (SystemConstant.TRACK_STAT_COMMENT.equals(mqVo.getStatType())) {
					albumStatMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_COMMENT, mqVo.getCount());
				}
			}
		} catch (Exception e) {
			//如果更新数据库发送异常，事务会进行回滚，下次再次投递消息允许继续处理统一个消息
			redisTemplate.delete(key);
			throw new RuntimeException(e);
		}
	}
}
